package com.kgcx.common.rabbitmq;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.ByteBufferInput;
import com.esotericsoftware.kryo.io.ByteBufferOutput;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.stereotype.Component;


/**
 * @author zhulong
 * kryo序列， 不支持 对象动态添加解析，慎用
 */
@Component
public class KryoMessageConverter extends AbstractMessageConverter {

    private static final Logger LOG = LoggerFactory.getLogger(KryoMessageConverter.class);

    public static final String CONTENT_TYPE = "application/x-kryo";
    public static final String DEFAULT_CHARSET = "UTF-8";
    private String defaultCharset = DEFAULT_CHARSET;
    private KryoFactory kryoFactory = new DefaultKryoFactory();

    @Override
    protected Message createMessage(Object object, MessageProperties messageProperties) {
        byte[] bytes = null;
        Kryo kryo = kryoFactory.create();
        Output output = new ByteBufferOutput(4096, 1024 * 1024);
        try {
            kryo.writeClassAndObject(output, object);
            bytes = output.toBytes();
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        } finally {
            output.close();
        }
        messageProperties.setContentType(CONTENT_TYPE);
        if (messageProperties.getContentEncoding() == null) {
            messageProperties.setContentEncoding(defaultCharset);
        }
        return new Message(bytes, messageProperties);
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        Object content = null;
        MessageProperties properties = message.getMessageProperties();
        if (properties != null) {
            if (properties.getContentType() != null && properties.getContentType().contains("x-kryo")) {
                Kryo kryo = kryoFactory.create();
                content = kryo.readClassAndObject(new ByteBufferInput(message.getBody()));
            } else {
                throw new MessageConversionException("Converter not applicable to this message");
            }
        }
        return content;
    }

    private class DefaultKryoFactory implements KryoFactory {

        @Override
        public Kryo create() {
            Kryo kryo = new Kryo();
            return kryo;
        }
    }
}
